package com.yw.mq.example.m04_transaction;

import cn.hutool.core.util.IdUtil;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import static com.yw.mq.core.MqConstant.DEFAULT_PRODUCER_GROUP;
import static com.yw.mq.core.MqConstant.NAME_SERVER_ADDR;

public class Producer {
    public static void main(String[] args) throws Exception {

        //  创建生产者
        TransactionMQProducer producer = new TransactionMQProducer(DEFAULT_PRODUCER_GROUP);
        // 设置连接信息
        producer.setNamesrvAddr(NAME_SERVER_ADDR);

        producer.setTransactionListener(new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                // 发送成功时(半消息）
                try {
                    // 业务代码
                    System.out.println("执行业务代码");
                    Thread.sleep(35 * 1000);
                    System.out.println("提交本地事务");
                    return LocalTransactionState.COMMIT_MESSAGE;
                } catch (Exception e) {
                    System.out.println("本地事务失败");
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                // 检测本地事务状态,必须检测完全失败才行，不然回滚后还是会提交
                System.out.println("检查本地事务，默认失败");
                return LocalTransactionState.ROLLBACK_MESSAGE;
            }
        });

        // 启动
        producer.start();

        Message message = new Message();
        message.setKeys(IdUtil.getSnowflakeNextIdStr());
        message.setTopic("order_topic");
        message.setTags("food");
        message.setBody("hello mq".getBytes());

        SendResult sendResult = producer.sendMessageInTransaction(message,null);

        System.out.println("发送结果：" + sendResult);
        
        producer.shutdown();

    }
}
